package monitorcon

import (
	"bytes"
	"errors"
	"github.com/streadway/amqp"
	"log"
	"time"
)

//ROUTINGKEY : mq的routing key
const ROUTINGKEY string = "monitordevice"

//QUEUE : mq 的队列名称
const QUEUE string = "goqueue"
//EXCHANGEMONITOR :队列exchange
const EXCHANGEMONITOR string = "monitor"

//rabbit mq
var channel *amqp.Channel
var hasMQ  = false
var conn *amqp.Connection

//SetupRMQ mq初始化 参数格式：amqp://用户名:密码@地址:端口号/host
func SetupRMQ(rmqAddr string) (err error) {
	var err1 error
	if channel == nil {
		conn, err1 = amqp.Dial(rmqAddr)
		if err != nil {
			return err
		}

		channel, err1 = conn.Channel()
		if err != nil {
			return err
		}

		err1 = Ping(QUEUE)

		if err1 == nil {
			hasMQ = true
		} else {
			hasMQ = false
		}

		return err1
	}
	return nil
}

//Ping  测试连接是否正常,并且创建exchange和队列
func Ping(queue string) error {

	if channel == nil {
		return errors.New("RabbitMQ is not initialize")
	}

	//func (ch *Channel) ExchangeDeclare(name, kind string, durable, autoDelete, internal, noWait bool, args Table) error

	err := channel.ExchangeDeclare(EXCHANGEMONITOR, "direct", true, false, false, true, nil)
	if err != nil {
		return err
	}
	log.Println("exchange creation succ")

	_, err = channel.QueueDeclare(
		QUEUE, // name
		true,  // durable
		true,  // delete when usused
		false, // exclusive
		false, // no-wait
		nil,   // arguments
	)

	if err != nil {
		return err
	}
	log.Println("channel creation succ")

	err = channel.QueueBind(
		QUEUE,           // queue name  队列的名字
		ROUTINGKEY,      // routing key  广播模式不需要这个
		EXCHANGEMONITOR, // exchange  交换机名字
		false,
		nil)

	if err != nil {
		return err
	}
	log.Println("bind succ")

	msg := amqp.Publishing{
		DeliveryMode: amqp.Persistent,
		Timestamp:    time.Now(),
		ContentType:  "text/plain",
		Body:         []byte("Go Go monitor!"),
	}

	//exchange         routingkey
	err = channel.Publish(EXCHANGEMONITOR, ROUTINGKEY, false, false, msg)

	if err != nil {
		return err
	}

	return err
}

//Close 关闭连接
func Close() {
	channel.Close()
	conn.Close()
	hasMQ = false
}

//Consumemessage 消费mq 信息
func Consumemessage(callback MQmsgCallback) {

	// Indicate we only want 1 message to acknowledge at a time.
	if err := channel.Qos(10, 0, false); err != nil {
		return
	}

	// Exclusive consumer                    (consumer name)
	deliveries, err := channel.Consume(QUEUE, "", false, false, false, false, nil)

	if err != nil {
		log.Println(err)
		return
	}

	//forever := make(chan bool) //不需要永久阻塞

	go func() {
		for msglog := range deliveries {
			// ... this consumer is responsible for sending pages per log
			s := BytesToString(&(msglog.Body))
			log.Println("msg:",*s)
			if callback(*s) {
				msglog.Ack(true)
			}
		}
	}()

	//log.Printf(" [*] Waiting for messages. To exit press CTRL+C\n")
	//<-forever

}

//BytesToString ...
func BytesToString(b *[]byte) *string {
	s := bytes.NewBuffer(*b)
	r := s.String()
	return &r
}
